package com.atguigu.gmall.realtime.joindemo;

import com.atguigu.gmall.realtime.app.BaseSQLApp;
import com.atguigu.gmall.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author lzc
 * @Date 2022/12/30 14:12
 */
public class LookupJoin extends BaseSQLApp {
    public static void main(String[] args) {
        new LookupJoin().init(
            6000,
            1,
            "LookupJoin"
        );
        
    }
    
    @Override
    public void handle(StreamExecutionEnvironment env,
                       StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table abc(" +
                            " id string," +
                            " pt as proctime() " + // 要用 lookup join 需要给这个维度表添加一个处理实际
                            ")" + SQLUtil.getKafkaSourceDDL("abc", "a", "csv"));
        
        // 通过jdbc 连接读取 mysql 中的字典表 
        tEnv.executeSql("create  table base_dic (" +
                            "  dic_code string," +
                            "  dic_name string " +
                            ") WITH (" +
                            "  'connector' = 'jdbc'," +
                            "  'url' = 'jdbc:mysql://hadoop162:3306/gmall2022?useSSL=false'," +
                            "  'table-name' = 'base_dic', " +
                            "  'username' = 'root', " +
                            // 对查到的维度数据, 缓存到内存中的时间
                            // 要在准确性和效率之间达到一个平衡
                            "  'lookup.cache.ttl' = '30 s', " +
                            // 对查到的维度数据, 最多缓存 100 条
                            "  'lookup.cache.max-rows' = '100', " +
                            "  'password' = 'aaaaaa' " +
                            ")");
        
        tEnv.sqlQuery("select " +
                          "abc.id, " +
                          "dic_name " +
                          "from abc " +
                          "join base_dic for system_time as of abc.pt as dic " +
                          "on abc.id=dic.dic_code ")
            .execute()
            .print();
    }
}
/*
id  user_id    下单
1     u_1      100      男
2     u_2      200		女
3     u_3      300      男

想统计 不同性别的金额的和

 事实表 join 维度表
 	普通的 join 不行, 只时候事实表与事实表的 join
 	
 	事实表可以做成流
 	
 	维度表无法做成流:
 	    维度表变化比较慢
 	    维度的每条数据都会重复使用
 	    
lookup join:
    用来专门解决事实表 join 维度表
 */